package org.kafka.connector;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;

public class MyTask extends SourceTask {

    private String filename;
    private String topic;
    private BufferedReader stream;
    private int lineNum;

    public String version() {
        return null;
    }

    public void start(Map<String, String> map) {
        //TODO 从offset中恢复
        lineNum = 0;
        this.filename = map.get(MyConnector.FILE_CONFIG);
        stream = getFileInputStrem(filename);
        this.topic = map.get(MyConnector.TOPIC);
    }

    public List<SourceRecord> poll() throws InterruptedException {
        ArrayList<SourceRecord> list = new ArrayList<>();
        if (null != stream) {
            while (list.isEmpty()) {
                String line = readLine(stream);
                if (null != line) {
                    lineNum++;
                    Map<String, String> sourcePartition = Collections.singletonMap("filename", filename);
                    Map<String, Integer> sourceOffset = Collections.singletonMap("position", lineNum);
                    list.add(new SourceRecord(sourcePartition, sourceOffset, topic, Schema.STRING_SCHEMA, line));
                }

            }
        }
        return list;
    }

    public void stop() {
        try {
            stream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private String readLine(BufferedReader bufferedReader) {
        try {
            return bufferedReader.readLine();
        } catch (IOException e) {
            return null;
        }
    }

    private BufferedReader getFileInputStrem(String filename) {
        try {
            return new BufferedReader(new FileReader(filename));
        } catch (FileNotFoundException e) {
            return null;
        }
    }
}
